package io.kiki.sba.registry.server.clientInterface.strategy.impl;

import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.store.Subscriber;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.core.constants.EventTypeConstants;
import io.kiki.sba.registry.core.model.RegisterResponse;
import io.kiki.sba.registry.core.model.SubscriberRegister;
import io.kiki.sba.registry.remoting.netty.BoltUtil;
import io.kiki.sba.registry.server.clientInterface.converter.SubscriberConverter;
import io.kiki.sba.registry.server.clientInterface.registry.Registry;
import io.kiki.sba.registry.server.clientInterface.strategy.SubscriberHandlerStrategy;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import static io.kiki.sba.registry.common.model.constants.ValueConstants.DEFAULT_INSTANCE_ID;


public class DefaultSubscriberHandlerStrategy implements SubscriberHandlerStrategy {
    private static final Logger logger = LoggerFactory.getLogger(DefaultSubscriberHandlerStrategy.class);

    @Autowired
    protected Registry sessionRegistry;

    @Override
    public void handleSubscriberRegister(Channel channel, SubscriberRegister subscriberRegister, RegisterResponse registerResponse) {
        Subscriber subscriber = null;

        try {
            String ip = channel.getRemoteAddress().getAddress().getHostAddress();
            int port = channel.getRemoteAddress().getPort();
            subscriberRegister.setIp(ip);
            subscriberRegister.setPort(port);

            if (StringUtils.isBlank(subscriberRegister.getZone())) {
                subscriberRegister.setZone(ValueConstants.DEFAULT_ZONE);
            }

            if (StringUtils.isBlank(subscriberRegister.getInstanceId())) {
                subscriberRegister.setInstanceId(DEFAULT_INSTANCE_ID);
            }

            subscriber = SubscriberConverter.convert(subscriberRegister);
            subscriber.setProcessId(ip + ":" + port);

            handle(subscriber, channel, subscriberRegister, registerResponse);
        } catch (Throwable e) {
            handleError(subscriberRegister, subscriber, registerResponse, e);
        }
    }

    protected void handle(Subscriber subscriber, Channel channel, SubscriberRegister subscriberRegister, RegisterResponse registerResponse) {
        subscriber.setClientUrl(new Url(channel.getRemoteAddress(), BoltUtil.getBoltCustomSerializer(channel)));
        subscriber.setServerUrl(new Url(channel.getLocalAddress()));

        final String eventType = subscriberRegister.getEventType();
        if (EventTypeConstants.REGISTER.equals(eventType)) {
            sessionRegistry.register(channel, subscriber);
        } else
            if (EventTypeConstants.UNREGISTER.equals(eventType)) {
                sessionRegistry.unRegister(subscriber);
            } else {
                logger.warn("unsupported subscriber.eventType:{}", eventType);
            }
        registerResponse.setVersion(subscriberRegister.getVersion());
        registerResponse.setRegistId(subscriberRegister.getRegistId());
        registerResponse.setSuccess(true);
        registerResponse.setMessage("Subscriber register success!");
        log(true, subscriberRegister, subscriber);
    }

    private void log(boolean success, SubscriberRegister subscriberRegister, Subscriber subscriber) {
        // [Y|N],[R|U|N],app,zone,dataInfoId,registerId,scope,elementType,clientVersion,clientIp,clientPort
        Metrics.Access.subCount(success);
        logger.info("{},{},{},{},{},G={},I={},{},{},{},{},{},{},{},{},pb={},attrs={},multi={}", success ? 'Y' : 'N', EventTypeConstants.getEventTypeFlag(subscriberRegister.getEventType()), subscriberRegister.getAppName(), subscriberRegister.getZone(), subscriberRegister.getDataId(), subscriberRegister.getGroup(), subscriberRegister.getInstanceId(), subscriberRegister.getRegistId(), subscriberRegister.getScope(), subscriber == null ? "" : subscriber.getElementType(), subscriber == null ? "" : subscriber.getClientVersion(), subscriber == null ? "" : subscriber.getRegisterTimestamp(), subscriber == null ? "" : subscriber.getVersion(), subscriberRegister.getIp(), subscriberRegister.getPort(), subscriber == null ? "0" : subscriber.attributesSize(), subscriberRegister.isAcceptMulti() ? 'Y' : 'N');
    }

    protected void handleError(SubscriberRegister subscriberRegister, Subscriber subscriber, RegisterResponse registerResponse, Throwable e) {
        log(false, subscriberRegister, subscriber);
        RegisterLogs.logError(subscriberRegister, "Subscriber", registerResponse, e);
    }
}
